package com.test.mq.producer;

import com.test.mq.mapper.TestMapper;
import com.lee.message.core.annotation.TransactionalMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
@Component
    public class RocketMqProducer {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Autowired
    private TestMapper testMapper;

    @Autowired
    private TestService testService;

    AtomicInteger ai = new AtomicInteger();

    /**
     * 发送普通消息
     */
    public void sendMsg(String msgBody) {
        log.info("发送消息: " + msgBody + ai.incrementAndGet());
        Message<String> message = MessageBuilder.withPayload(msgBody).setHeader(MessageConst.PROPERTY_KEYS, UUID.randomUUID()).build();
        rocketMQTemplate.syncSend("queue_test_topic", message);
    }

    /**
     * 发送事务消息
     * 注解TransactionalMessage必须是事务的开始
     */
    @TransactionalMessage
    public void sendMessageInTransaction(String msgBody) {
        log.info("发送消息: " + msgBody + ai.incrementAndGet());
        Map<String, String> params = new HashMap<>();
        params.put("id", UUID.randomUUID().toString());
        params.put("name", msgBody + ai.incrementAndGet());
        testMapper.insert(params);
        testService.test1(msgBody);
        testService.test2(msgBody);
        Message<String> message = MessageBuilder.withPayload(msgBody + ai.incrementAndGet()).setHeader(MessageConst.PROPERTY_KEYS, UUID.randomUUID()).build();
        /**
         * 注：发送逻辑后面必须没有事务操作，sendMessageInTransaction会提交当前事务，后面的操作将不带事务
         * 建议必须把sendMessageInTransaction放整个业务代码最后面
         */
        rocketMQTemplate.sendMessageInTransaction("queue_test_topic", message, null);
    }

    /**
     * 发送延迟消息
     * <p>
     * 就是消息发送后消费端要过一段时间才能收到
     * <p>
     * 延迟的级别
     * "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";
     * <p>
     * 原理:
     * 消息是在broker上的但是状态是不可消费的，只有到了延迟的结束时间，才可以被消费者消费
     */
    public void sendDelayMessage(String msgBody) {
        log.info("发送消息: " + msgBody + ai.incrementAndGet());
        Message<String> message = MessageBuilder.withPayload(msgBody + ai.incrementAndGet()).
                setHeader(MessageConst.PROPERTY_KEYS, UUID.randomUUID()).
                // 设置延迟级别，级别参考方法注释
                setHeader(MessageConst.PROPERTY_DELAY_TIME_LEVEL, 5).
                build();

        rocketMQTemplate.syncSend("queue_test_topic", message);
    }

    /**
     * 发送一次消息
     * 发送速度很快但是没有保障的发送方式
     * 无需等待
     * 主要可以用来做日志采集，适用于某些耗时非常短，但是对可靠性要求不高的场景
     */
    public void sendOneWay(String msgBody) {
        log.info("发送消息: " + msgBody + ai.incrementAndGet());
        Message<String> message = MessageBuilder.withPayload(msgBody + ai.incrementAndGet()).
                setHeader(MessageConst.PROPERTY_KEYS, UUID.randomUUID()).
                build();

        rocketMQTemplate.sendOneWay("queue_test_topic", message);
    }

    /**
     * 发送排序消息
     */
    public void syncSendOrderly(String msgBody) {
        log.info("发送消息: " + msgBody + ai.incrementAndGet());
        Message<String> message = MessageBuilder.withPayload(msgBody + ai.incrementAndGet()).
                setHeader(MessageConst.PROPERTY_KEYS, UUID.randomUUID()).
                build();

        // 方法主要多了一个hashKey，用于保证hashKey一样的消息发送同一个队列
        rocketMQTemplate.syncSendOrderly("queue_test_topic", message, ai.get()+"");
    }

}
